package vip.xiaomaoxiaoke.msg.service;


import org.apache.commons.collections4.CollectionUtils;
import vip.xiaomaoxiaoke.msg.interceptor.LocalMessageAnnotationInterceptor;
import vip.xiaomaoxiaoke.msg.message.Message;
import vip.xiaomaoxiaoke.msg.message.impl.NativeMethodMessage;
import vip.xiaomaoxiaoke.msg.repository.LocalMessageRepository;
import vip.xiaomaoxiaoke.msg.repository.LocalMessageTable;
import vip.xiaomaoxiaoke.msg.strategies.RefusalStrategies;
import vip.xiaomaoxiaoke.msg.util.SerializeUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.lang.reflect.InvocationTargetException;
import java.util.Comparator;
import java.util.List;

/**
* 可靠方法执行服务
 */
@Slf4j
public class ReliableMethodExecuteService {


    /**
     * 本地消息数据源
     * */
    private LocalMessageRepository localMessageRepository;


    /**
     * 拒绝策略
     * */
    private RefusalStrategies refusalStrategies ;


    public ReliableMethodExecuteService(LocalMessageRepository localMessageRepository, RefusalStrategies refusalStrategies) {
        this.localMessageRepository = localMessageRepository;
        this.refusalStrategies = refusalStrategies;
    }


    /**
     * 加载并发送，建议：分布式环境使用xxljob执行，
     * 单机环境使用本地定时任务执行
     * */
    public void loadAndResend(int sizePreTask) {
        Long  latestId = 0L;
        List<LocalMessageTable> localMessageTables = this.localMessageRepository.loadNotSuccessByUpdateGt(latestId, sizePreTask);
        while (CollectionUtils.isNotEmpty(localMessageTables)) {

//            log.info("消息重发");
            retrySend(localMessageTables);

            latestId = localMessageTables.stream().map(LocalMessageTable::getId).max(Comparator.naturalOrder()).orElse(Long.MAX_VALUE);


            localMessageTables = this.localMessageRepository.loadNotSuccessByUpdateGt(latestId, sizePreTask);
        }

    }

    /**
     * 重新发送
     * */
    private void retrySend(List<LocalMessageTable> localMessageTables) {

        localMessageTables.stream().forEach(localMessageTable -> {
            try {
                run(localMessageTable);
            } catch (InvocationTargetException e) {
                throw new RuntimeException(e);
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 保存并发送
     * */
    public void saveAndSend(Message message) {
        LocalMessageTable localMessageTable = message.convert();

        localMessageTable.init();
        saveToDB(localMessageTable);
        try {
            addCallbackOrRunTask(localMessageTable);
        } catch (InvocationTargetException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }

    }

    private void addCallbackOrRunTask(LocalMessageTable localMessageTable) throws InvocationTargetException, IllegalAccessException {
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            // 添加监听器，在事务提交后触发后续任务
            TransactionSynchronization transactionSynchronization = new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    try {
                        run(localMessageTable);
                    } catch (InvocationTargetException e) {
                        throw new RuntimeException(e);
                    } catch (IllegalAccessException e) {
                        throw new RuntimeException(e);
                    }
                }
            };
            TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);

        } else {
            run(localMessageTable);
        }
    }


    public void run(LocalMessageTable localMessageTable) throws InvocationTargetException, IllegalAccessException {
        // 判断是否需要执行拒绝策略
        if (localMessageTable.getRetryCount() >= localMessageTable.getMaxRetries()) {
            try {
                refusalStrategies.execute(localMessageTable);
                localMessageTable.setRefusalStrategiesState(1);
                localMessageRepository.update(localMessageTable);
            } catch (Exception e) {
                Throwable cause = e.getCause();
                if (e.getCause()!=null) {
                    localMessageTable.setResponseContent(cause.getMessage());
                } else {
                    localMessageTable.setResponseContent(e.getMessage());
                }
                localMessageTable.setRefusalStrategiesState(-1);
                localMessageRepository.update(localMessageTable);
            }
            return;
        }


        NativeMethodMessage convert = new NativeMethodMessage().convert(localMessageTable);

        LocalMessageAnnotationInterceptor.Mk mk = LocalMessageAnnotationInterceptor.map.get(convert.getMethodKey());
        try {
            Object invoke = mk.getMethod().invoke(mk.getBean(), SerializeUtil.deserialize(convert.getParamValueList(), mk.getMethod()));
            // 没有异常则执行成功，设置执行状态为成功
            localMessageTable.setState(1);

            // 保持方法返回结果
            localMessageTable.setResponseContent(JSON.toJSONString(invoke));
            // 更新数据库结果
            localMessageRepository.update(localMessageTable);
        } catch (Exception e) {
            e.printStackTrace();
            // 执行发生异常
            // 记录异常信息
            log.error(e.getMessage());
            Throwable cause = e.getCause();
            if (e.getCause()!=null) {
                localMessageTable.setResponseContent(cause.getMessage());
            } else {
                localMessageTable.setResponseContent(e.getMessage());
            }

            // 设置执行状态为失败
            localMessageTable.setState(-1);
            // 设置重试次数
            localMessageTable.setRetryCount(localMessageTable.getRetryCount() + 1);
            localMessageRepository.update(localMessageTable);
        }


    }



    /**
     * 保存消息
     */
    private void saveToDB(LocalMessageTable localMessage) {
        this.localMessageRepository.save(localMessage);
    }


}
